package pageRank;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.Arrays;
import java.util.Iterator;

import mapred.AbstractReducer;
import mapred.util.Bytes;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import removeDanglingPage.ReadBytes;

public class PageRank_Reducer extends AbstractReducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable> {

	@Override
	public void reduce(BytesWritable key, Iterator<BytesWritable> values,
			OutputCollector<BytesWritable, BytesWritable> output,
			Reporter reporter) throws IOException {
		// TODO Auto-generated method stub
		double old = 0;
		double pagerank = 0;
		byte[] link = {};
		byte[] Z = new byte[16];
		Arrays.fill(Z, (byte)0);
		while(values.hasNext()) {
			ReadBytes v = new ReadBytes(values.next());
			if(v.size() == 8) {
				pagerank += v.nextDouble();
			} else if(v.size() == 9) {
				v.next();
				old = v.nextDouble();
				reporter.incrCounter("PageRank", "Old", 1);
			} else {
				link = v.nextBytes(v.remain());
			}
		}
		ReadBytes k = new ReadBytes(key);
		if(Arrays.equals(Z, k.nextBytes(k.remain()))) {
			BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(
					FileSystem.get(this.getJobConf()).create(
							new Path("Z_" + this.getJobConf().get("Round")),
							true)));
			bw.write(String.valueOf(pagerank));
			bw.flush();
			bw.close();
		} else {
			if (pagerank < 0) {
				pagerank = 1;
			} else {
				pagerank = pagerank * 0.85 + 0.15;
			}
			output.collect(key, new BytesWritable(Bytes.concat(Bytes.Double2Bytes(pagerank), link)));
			reporter.incrCounter("PageRank", "WebSize", 1);
			if (Math.abs(pagerank - old) > 1e-6) {
				reporter.incrCounter("PageRank", "Delta", 1);
			}
		}
	}

//	@Override
//	public void reduce(Text key, Iterator<Text> values,
//			OutputCollector<Text, Text> output, Reporter reporter)
//			throws IOException {
//		double old = 0;
//		double pagerank = 0;
//		String link = "";
//		while (values.hasNext()) {
//			String value = values.next().toString();
//			if (value.startsWith("#")) {
//				pagerank += Double.parseDouble(value.substring(1));
//			} else if (value.startsWith("@")){
//				old = Double.parseDouble(value.substring(1));
//			} else {
//				link = value;
//			}
//		}
//		if ("Z".equals(key.toString())) {
//			BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(
//					FileSystem.get(this.getJobConf()).create(
//							new Path("Z_" + this.getJobConf().get("Round")),
//							true)));
//			bw.write(String.valueOf(pagerank));
//			bw.flush();
//			bw.close();
//		} else {
//			if (pagerank < 0) {
//				pagerank = 1;
//			} else {
//				pagerank = pagerank * 0.85 + 0.15;
//			}
//			output.collect(key, new Text(pagerank + "\t" + link));
//			reporter.incrCounter("PageRank", "WebSize", 1);
//			if (Math.abs(pagerank - old) > 1e-6) {
//				reporter.incrCounter("PageRank", "Delta", 1);
//			}
//		}
//	}

}
